package com.atguigu.api4

import com.atguigu.api.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}

/**
 * @description: 流式输出
 * @time: 2020/7/22 17:22
 * @author: baojinlong
 **/
object OutputTableTest {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    // 从文本读取
    val inputStreamFromFile: DataStream[String] = environment.readTextFile("E:/qj_codes/big-data/FlinkTutorial/src/main/resources/sensor.data")


    // 基本转换操作
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      })

    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useOldPlanner()
      .inStreamingMode()
      .build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(environment, settings)

    // 将DataStream转成Table
    val sensorTable: Table = tableEnv.fromDataStream(dataStream, 'id, 'temperature as 'temp, 'timestamp as 'ts)
    // 对table进行操作得到结果表
    val resultTable: Table = sensorTable
      .select('id, 'temp)
      .filter('id === "sensor_01")
    // 聚合操作得到表数据
    val aggResultTable: Table = sensorTable
      .groupBy('id)
      .select('id, 'id.count as 'cnt)

    // 定义一张输出表,这就是要写入数据的TableSink
    tableEnv.connect(new FileSystem().path("E:/qj_codes/big-data/FlinkTutorial/src/main/resources/sensor-out33.data"))
      .withFormat(new Csv)
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("temp", DataTypes.DOUBLE())
      )
      .createTemporaryTable("outputTableTest")

    // 把结果写入到表中tableSink,输出文件只能是append
    resultTable.insertInto("outputTableTest")
    // 注意aggResultTable不能写入到文件中.

    //sensorTable.printSchema()
    sensorTable.toAppendStream[(String, Double, Long)].print("datastream2TableDemo")

    // 执行
    environment.execute("OutputTableTest job")
  }

}
